package com.example.parsing;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.example.parsing.listener.MyListener;
import com.example.parsing.producer.MyProducer;
import com.example.parsing.unit.Esn;
import com.example.parsing.unit.NoGenerator;
import com.example.parsing.unit.StringReplace;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.MessagingException;
import org.springframework.stereotype.Component;

import java.io.BufferedReader;
import java.io.File;
import java.io.FileReader;
import java.util.*;

@Component
class JsonMessageHandler implements MessageHandler {
	
	private static final Logger logger = LoggerFactory.getLogger(JsonMessageHandler.class);
	
	private static final Map<String, Date> fileMap = new HashMap<>(); // 保存已经执行过的文件
	
	private static final List<String> numList = new ArrayList<>(); // 保存已经进行注册的设备
	
	private static final Map<String, String> configMap = ConfigProperties.readConfig(); //读取配置文件的属性，转换成k-v形式
	
	private final KafkaTemplate<String, String> kafkaTemplate;
	
	private final MyProducer myProducer;
	
	@Value("${myAddress.DEVICETYPE}")
	private String DEVICETYPE; // 设备名称
	
	public JsonMessageHandler(KafkaTemplate<String, String> kafkaTemplate, MyProducer myProducer) {
		this.kafkaTemplate = kafkaTemplate;
		this.myProducer = myProducer;
	}
	
	@Override
	public void handleMessage(Message<?> message) throws MessagingException {
		logger.info("------------------handler处理消息----------------");
		try {
			File file = (File) message.getPayload();
			String path = file.toString();
			BufferedReader br = new BufferedReader(new FileReader(path));
            MessageHeaders s = message.getHeaders();
            String fileName = (String) s.get("file_name");
            JSONObject jsonObject = (JSONObject) JSON.parse(br.readLine());
            if (!fileMap.containsKey(fileName)) {
                for (Map.Entry<String, Object> entry : jsonObject.entrySet()) {
	                JSONObject jsonObject1 = (JSONObject) entry.getValue();
	                String empiricalName = null;
	                assert configMap != null;
	                for (Map.Entry<String, String> entry1 : configMap.entrySet()) {
		                if (entry1.getValue().equals("empiricalName")) {
			                empiricalName = entry1.getKey();
		                }
	                }
	                String empiricalNameValue = (String) jsonObject1.get(empiricalName);
	                if (!numList.contains(empiricalNameValue)) {
		                // 如果能进来就说明未注册，进行注册
		                numList.add(empiricalNameValue);
		                register(jsonObject1, empiricalNameValue, fileName);
	                } else {
		                // 如果进到这个分支说明设备已经注册过了，进行上报
		                report(jsonObject1, empiricalNameValue, fileName);
	                }
                }
            }
        } catch (Exception e) {
            logger.error("出错了，出错的类型是=" + e);
        }
    }

    public void register(JSONObject jsonObject, String deviceName, String fileName) {
        String esn = Esn.createID();
        try {
            deviceName = StringReplace.replace(deviceName); // 将deviceName替换成文档所需要求的
            String deviceTitle = deviceName + "实验数据";
            myProducer.register(esn, DEVICETYPE, deviceName, deviceTitle, kafkaTemplate);
            report(jsonObject, deviceName, fileName);
        } catch (Exception e) {
            logger.error("数据上报失败，错误消息是" + e);
        }
    }

    public void report(JSONObject jsonObject, String deviceName, String fileName) {
        try {
	        JSONObject jsonObject1 = new JSONObject();
	        assert configMap != null;
	        for (String key : jsonObject.keySet()) {
		        if (configMap.containsKey(key)) {
			        jsonObject1.put(configMap.get(key), jsonObject.get(key));
		        }
	        }
	        JSONArray jsonArray = new JSONArray();
	        jsonArray.add(jsonObject1);
	        JSONObject json = new JSONObject();
	        JSONObject requestJson = new JSONObject();
	        json.put("id", NoGenerator.getUUID());
	        json.put("requestTime", String.valueOf(NoGenerator.getDate()));
	        deviceName = StringReplace.replace(deviceName);
	        logger.info("-----------------------deviceName=" + deviceName);
	        String deviceId = MyListener.deviceName.get(deviceName);
	        json.put("deviceId", deviceId);
	        json.put("modelName", DEVICETYPE);
	        requestJson.put("serviceProperties", jsonArray);
	        json.put("request", requestJson);
	        logger.info("数据开始上报，上报的数据是" + json);
	        kafkaTemplate.send(
				        "sys." + DEVICETYPE + ".thing.serviceid.property.post", DEVICETYPE, json.toJSONString());
	        if (deviceId != null) {
		        fileMap.put(fileName, new Date());
	        }
        } catch (Exception e) {
            logger.error("数据上报失败，错误消息是" + e);
        }
    }
}
